package raft

import "sort"

func max(a, b int) int{
	if a > b {
		return a
	}
	return b
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

// raft 日志复制管理
// 日志结构
type  LogEntry struct {
	Term		int				// 任期号
	Command		interface{}	// client端发送的命令
}

// 日志请求结构
type AppendEntriesArgs struct {
	Term			int			// leader任期号
	// 在raft中，有可能会出现client直接连上follower,此时,follower需要给client
	// 发送leaderId，方便follower重定向
	LeaderId		int			// leader ID so follower can redirect clients
	PrevLogIndex	int			// 新的日志条目紧随之前的索引值
	PrevLogTerm		int			// PrevLogIndex任期编号
	Entries			[]LogEntry	// 准备存储的日志条目
	LeaderCommit	int			// leader已经提交的日志索引
}

// 日志复制响应结构
type AppendEntriesReply struct {
	CurrentTerm int // 当前任期号，主要用于leader更新自己
	// 跟随者包含了匹配上 prevLogIndex
	// 和 prevLogTerm 的日志时为真
	Success bool

	// 自定义冲突相关变量
	ConflictTerm int // 冲突日志的任期编号
	FirstIndex   int // 存储第一个冲突编号的日志索引
}

// 应用日志进程 
func (rf *Raft) applyEntryDaemon()  {
	// 日志提交完成之后
	// 将日志进行应用，返回...
	for {
		var logs []LogEntry
		rf.mu.Lock()
		// 判断，如果节点最后的应用日志索引与已提交的日志索引相等
		// 说明提交过的所有日志都已经被应用
		for rf.lastApplied == rf.commitIndex {
			rf.commitCond.Wait()
			select {
			// 检测是否有中断指令
			case <-rf.shutdown:
				rf.mu.Unlock()
				close(rf.applyCh)
				return
			default:
			}
		}
		// 获取最后应用的日志索引与最新提交的日志索引
		last, cur := rf.lastApplied, rf.commitIndex
		if last < cur {
			// 满足该条件说明还有一部分已经被提交的日志没有被应用
			rf.lastApplied = rf.commitIndex
			// 找到已经提交但未更新的这一部分日志
			logs = make([]LogEntry, cur - last)
			copy(logs, rf.Logs[last + 1:cur + 1])
		}
		rf.mu.Unlock()
		// 对还没有被应用的日志进行应用
		for i := 0 ; i < cur - last; i++ {
			reply := ApplyMsg{
				Index: last + i + 1,
				Command: logs[i].Command,
			}
			// 传回响应
			rf.applyCh <- reply
		}
	}
}

// 唤醒一致性检查
func (rf *Raft) wakeupConsistencyCheck() {
	for i := 0; i < len(rf.peers); i++ {
		if i != rf.me {
			rf.newEntryCond[i].Broadcast()
		}
	}
}

// 启动日志复制进程
func (rf *Raft) logEntryAgreeDaemon()  {
	// 遍历节点，向其它每个节点发起日志复制操作
	for i := 0; i < len(rf.peers); i++ {
		if i != rf.me {
			go rf.consistencyCheckDaemon(i)
		}
	}
}

// 发起日志复制操作
func (rf *Raft)consistencyCheckDaemon(n int)  {
	for {
		rf.mu.Lock()
		// 每个节点都在等待client提交命令到leader上去
		rf.newEntryCond[n].Wait()
		select {
		case <- rf.shutdown:
			rf.mu.Unlock()
			return
		default:
		}

		// 判断节点角色，只有leader才能发起日志复制
		if rf.isLeader {
			var args AppendEntriesArgs
			args.Term = rf.CurrentTerm
			args.LeaderId = rf.me
			args.LeaderCommit = rf.commitIndex
			args.PrevLogIndex = rf.nextIndex[n] - 1
			args.PrevLogTerm = rf.Logs[args.PrevLogIndex].Term
			// 判断是否有新的日志进来
			// len(rf.Logs) : leader当前的日志总数
			// rf.nextIndex[n]:leader要发送给节点n的下一个日志索引
			// leader的日志长度大于leader所知道的follow n的日志长度
			if rf.nextIndex[n] < len(rf.Logs) {
				// 添加新的日志
				args.Entries = append(args.Entries, rf.Logs[rf.nextIndex[n]:]...)
			} else {
				args.Entries = nil
			}
			rf.mu.Unlock()
			// 新建响应的通道，在发起日志复制请求之后，存储结果
			replyCh := make(chan AppendEntriesReply, 1)
			go func() {
				// 日志复制的响应内容
				var reply AppendEntriesReply
				// 发起日志复制请求
				if rf.sendAppendEntries(n, &args, &reply) {
					replyCh <- reply
				}
			}()

			// 获取响应
			select {
			case reply := <-replyCh:
				rf.mu.Lock()
				if reply.Success {
					// 响应成功
					rf.matchIndex[n] = reply.FirstIndex //FirstIndex是第一个发成冲突的下标，故FirstIndex前都是正确的日志
					rf.nextIndex[n] = rf.matchIndex[n] + 1

					// 提交日志(更新已提交的日志索引)
					rf.updateCommitIndex()
				} else {
					// 响应失败
					// 判断响应传回的term与当前节点(leader)的term谁大
					if reply.CurrentTerm > args.Term {
						//当前节点的任期号
						//TODO 首先需要判断是否需要更新当前节点的任期号
						if reply.CurrentTerm>rf.CurrentTerm{
							// 更新当前leader的任期号
							rf.CurrentTerm = reply.CurrentTerm
							rf.VotedFor = -1
						}
						// leader接收的是每个节点返回的响应
						// 所以在进行当前判断的时候，可能角色状态已经修改过
						if rf.isLeader {
							rf.isLeader = false
							// 变更角色状态，发起一致性检查
							rf.wakeupConsistencyCheck()
						}
						//持久化
						rf.persist()
						rf.mu.Unlock()
						rf.resetTimer <- struct{}{}
						return
					}

					// 解决冲突相关问题
					// know:当前leader能否找到冲突
					// lastIndex:代表节点中最后一个包含冲突任期编号的日志索引
					var know, lastIndex = false, 0
					if reply.ConflictTerm != 0 {
						// 查找最后一个包含冲突的日志索引
						for i := len(rf.Logs) - 1; i > 0; i-- {
							// 找到产生冲突的任期编号
							if rf.Logs[i].Term == reply.ConflictTerm {
								know = true
								lastIndex = i
								break
							}
						}

						// 如果找到了冲突编号
						if know {
							// 判断当前获取到的冲突编号索引与响应中的第一个冲突日志索引的大小
							if lastIndex > reply.FirstIndex {
								// 满足当前条件的情况下，说明
								// 在最后一个产生之前，已经有了一个冲突编号
								// 只允许存在第一个
								lastIndex = reply.FirstIndex
							}
							rf.nextIndex[n] = lastIndex
						}else {
							rf.nextIndex[n] = reply.FirstIndex
						}
					} else {
						// 如果响应中没有返回冲突的任期编号
						rf.nextIndex[n] = reply.FirstIndex
					}
					// 1 <= rf.nextIndex[n] <= len(rf.Logs)
					rf.nextIndex[n] = min(max(rf.nextIndex[n],1), len(rf.Logs))
				}
				rf.mu.Unlock()
			}
		} else {
			// 当前节点不是leader，直接返回
			rf.mu.Unlock()
			return
		}
	}
}

// 发起日志复制的请求
func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

// 更新日志提交索引
/*	TODO 如果存在一个满足N > commitIndex的 N，
	TODO 并且大多数的matchIndex[i] ≥ N成立，
	TODO 并且log[N].term == currentTerm成立，
	TODO 那么令 commitIndex 等于这个 N */
func (rf *Raft) updateCommitIndex()  {
	// 对于每一个follower，需要提交的日志索引
	// 应该是目前复制给leader的最高的日志索引
	match := make([]int, len(rf.matchIndex))
	copy(match, rf.matchIndex)
	// 排序
	sort.Ints(match)
	// 取一个满足条件的变量"N"
	target := match[len(rf.peers) / 2]
	// 为什么要取中间值，因为经过排序之后，索引是从小到大的
	// 取中间值就可以满足"大多数的matchIndex[i] ≥ N成立"这个条件
	if rf.commitIndex < target {
		if rf.Logs[target].Term == rf.CurrentTerm {
			rf.commitIndex = target
			go func() {rf.commitCond.Broadcast()}()
		}
	}
}

// 接收到日志复制请求之后的处理
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	select {
	case <-rf.shutdown:
		return
	default:
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	// Reply false if term < currentTerm
	if args.Term < rf.CurrentTerm {
		// 将比leader更新的任期填充到响应中，方便leader接收之后进行更新
		reply.CurrentTerm = rf.CurrentTerm
		reply.Success = false
		return
	}

	// 如果当前角色是Leader，更改节点角色(强一致性)
	if rf.isLeader {
		rf.isLeader = false
		// 唤醒一致性检查
		rf.wakeupConsistencyCheck()
	}

	// 如果不相等，那么在出现client直接访问当前follower的时候
	// follower会重定向到一个错误的节点
	if rf.VotedFor != args.LeaderId {
		rf.VotedFor = args.LeaderId
	}

	// 当前节点任期编号小于leader，覆盖
	if rf.CurrentTerm < args.Term {
		rf.CurrentTerm = args.Term
	}

	// 重置
	rf.resetTimer <- struct{}{}

	/*
		TODO 在发送附加日志 RPC 的时候，leader会把新的日志条目
		todo 紧接着之前的条目的索引位置(prevLogIndex)和
		TODO 任期号(prevLogTerm)包含在里面。
		TODO 如果跟随者在它的日志中找不到包含相同索引位置和任期号的条目，
		TODO 那么他就会拒绝接收新的日志条目
	*/
	preLogIdx, preLogTerm := 0, 0
	// 大于说明在follower的日志中能够找到args包含的prevLogIndex
	// 包含相同索引位置
	if len(rf.Logs) > args.PrevLogIndex {
		preLogIdx = args.PrevLogIndex
		// 从follower节点的日志中获取到的与args传入的索引相同的任期编号
		preLogTerm = rf.Logs[preLogIdx].Term
	}

	// 判断是否匹配
	if preLogIdx == args.PrevLogIndex &&
		// 包含相同任期号
	preLogTerm == args.PrevLogTerm {
			reply.Success = true
			// 截取当前已知的最后一个匹配，将不匹配的丢掉
			rf.Logs = rf.Logs[:preLogIdx+1]
			// 追加日志
			rf.Logs = append(rf.Logs, args.Entries...)
			// 获取到更新之后的最后一个日志索引
			var last = len(rf.Logs) - 1
			// 更新一下commitIndex
			// 如果 leaderCommit > commitIndex，
			// 令 commitIndex 等于 leaderCommit 和 新日志条目索引值中较小的一个
			if args.LeaderCommit > rf.commitIndex {
				rf.commitIndex = min(args.LeaderCommit, last)
				// commitIndex更新
				go func() {rf.commitCond.Broadcast()}()
			}
			// 更新最后一个冲突日志
			reply.ConflictTerm = rf.Logs[last].Term
			reply.FirstIndex = last
	} else {
		reply.Success = false
		// 处理冲突任期编号
		var first = 1
		reply.ConflictTerm = preLogTerm
		// 如果说新的conflictTerm= 0
		// 意味着leader有更多的日志需要复制或者是follower没有日志
		if reply.ConflictTerm == 0 {
			first = len(rf.Logs)
			// 将响应的冲突日志条目设置为当前节点最后一个日志索引条目
			reply.ConflictTerm = rf.Logs[first - 1].Term
		} else {
			// 不为0说明任期编号有冲突
			// 也就是说preLogTerm != args.PrevLogTerm
			for i := preLogIdx - 1; i > 0; i-- {
				if rf.Logs[i].Term != preLogTerm {
					// 遍历查找产生冲突的日志
					first = i + 1
					break
				}
			}
		}
		// 第一个产生冲突的日志
		reply.FirstIndex = first
	}
	//持久化
	rf.persist()
}